package com.xl.flinkdemo.TableAPI;

import com.xl.flinkdemo.entity.SensorReading;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

/**
 * @ClassName Example
 * @Description TODO
 * @Author hxl
 * @Date 2021/5/21 11:37
 * Version 1.0
 **/
public class ExampleKafkaPipeLine {
  public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
    environment.setParallelism(1);
    EnvironmentSettings oldStreamSettings = EnvironmentSettings.newInstance()
        .inStreamingMode()
        .useOldPlanner()
        .build();
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(environment, oldStreamSettings);

    //开始进行tableAPI
    tableEnv.connect(new Kafka()
        .version("0.11")
        .topic("testInput")
        .property("zookeeper.connect", "localhost:2181")
        .property("bootstrap.servers", "localhost:9092")
    ).withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("sensors", DataTypes.DOUBLE())).createTemporaryTable("inputTable");


    Table source = tableEnv.from("inputTable");
    Table resultTable = source.select("id,sensors").where(" id = 'sensor_4'");

    tableEnv.connect(new Kafka()
         .version("0.11")
            .topic("test")
            .property("zookeeper.connect", "localhost:2181")
            .property("bootstrap.servers", "localhost:9092")
    ).withFormat(new Csv())
        .withSchema(new Schema()
            .field("id", DataTypes.STRING())
            .field("sensors", DataTypes.DOUBLE())).createTemporaryTable("outTable");

    resultTable.insertInto("outTable");

    environment.execute();

  }
}
